package main.scala.org.huel.dataprocessing

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

/**
 * @author Liweijian
 * @Description 使用Spark Core/RDD API统计各班级学生出勤和缺勤数量
 * attendance_summary  总体缺勤统计
 * absence_prediction   可能缺勤学生
 * @date 2025/06/13
 */
object ClassAttendanceRddAnalysis {

  // 定义考勤记录的case class
  case class AttendanceRecord(
                               classId: String,
                               studentName: String,
                               courseName: String,
                               studentId: String,
                               score: Int,
                               isAbsent: Int
                             )

  // 定义班级出勤统计的case class
  case class ClassAttendanceSummary(
                                     classId: String,
                                     presentCount: Int,
                                     absentCount: Int,
                                     totalCount: Int,
                                     attendanceRate: Double
                                   )

  def main(args: Array[String]): Unit = {
    /*
    * 配置spark
    * */
    // 创建SparkConf
    val sparkConf = new SparkConf()
      .setAppName("ClassAttendanceRddAnalysis")
      .setMaster("local[*]")  // 本地运行模式

    // 创建StreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 获取SparkContext
    val sc = ssc.sparkContext

    // 设置日志级别
    sc.setLogLevel("WARN")

    // Kafka参数配置
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "43.143.125.94:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "class_attendance_rdd_group",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    // 订阅Kafka主题
    val topics = Array("attendance")

    // 打印启动信息
    println(s"\n✓ 正在监听Kafka主题: ${topics.mkString(", ")}")
    println(s"✓ 使用Kafka服务器: 43.143.125.94:9092")
    println(s"✓ 消费者组ID: class_attendance_rdd_group")

    /*
    * 处理
    * */
    // 创建直接流
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    // 处理每批次RDD数据
    stream.foreachRDD { rdd =>
      // 获取偏移量范围以便稍后提交
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      println(s"收到RDD，包含 ${rdd.count()} 条记录")

      if (!rdd.isEmpty()) {
        // 获取当前时间
        val currentTime = LocalDateTime.now()
        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val formattedTime = currentTime.format(formatter)

        println(s"\n=========================================================")
        println(s"| 班级出勤统计报告 (RDD API) - ${formattedTime} |")
        println(s"=========================================================")

        // 1. 将kafka消息值转换为AttendanceRecord对象
        val attendanceRecordsRDD = rdd.map(record => record.value())
          .map { line =>
            val fields = line.split("\t")
            if (fields.length == 6) {
              AttendanceRecord(
                classId = fields(0),
                studentName = fields(1),
                courseName = fields(2),
                studentId = fields(3),
                score = fields(4).toInt,
                isAbsent = fields(5).toInt
              )
            } else {
              // 处理格式不正确的数据
              AttendanceRecord("Unknown", "Unknown", "Unknown", "Unknown", -1, -1)
            }
          }
          .filter(record => record.score >= 0) // 过滤掉无效记录

        // 2. 按班级分组统计出勤和缺勤数量
        val classAttendanceRDD = attendanceRecordsRDD
          .map(record => (record.classId, record.isAbsent))
          .groupByKey()
          .mapValues { absenceValues =>
            val valuesList = absenceValues.toList
            val totalCount = valuesList.size
            val absentCount = valuesList.count(_ == 1)
            val presentCount = totalCount - absentCount
            val attendanceRate = if (totalCount > 0) (presentCount.toDouble / totalCount) * 100 else 0.0
            (presentCount, absentCount, totalCount, attendanceRate)
          }
          .map { case (classId, (present, absent, total, rate)) =>
            ClassAttendanceSummary(
              classId = classId,
              presentCount = present,
              absentCount = absent,
              totalCount = total,
              attendanceRate = BigDecimal(rate).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble
            )
          }
          .sortBy(_.classId)

        // 3. 计算总体出勤统计
        val overallStats = attendanceRecordsRDD
          .map(record => record.isAbsent)
          .aggregate((0, 0))( // (presentCount, absentCount)
            (acc, value) => value match {
              case 0 => (acc._1 + 1, acc._2)     // 出勤
              case 1 => (acc._1, acc._2 + 1)     // 缺勤
              case _ => acc                      // 无效值
            },
            (acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
          )

        val totalPresent = overallStats._1
        val totalAbsent = overallStats._2
        val totalCount = totalPresent + totalAbsent
        val overallAttendanceRate = if (totalCount > 0)
          BigDecimal((totalPresent.toDouble / totalCount) * 100).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble
        else 0.0

        // 4. 课程与班级交叉统计
        val coursesPerClassRDD = attendanceRecordsRDD
          .map(record => ((record.courseName, record.classId), record.isAbsent))
          .groupByKey()
          .mapValues { absenceValues =>
            val valuesList = absenceValues.toList
            val totalCount = valuesList.size
            val absentCount = valuesList.count(_ == 1)
            val presentCount = totalCount - absentCount
            val attendanceRate = if (totalCount > 0) (presentCount.toDouble / totalCount) * 100 else 0.0
            (presentCount, absentCount, totalCount, attendanceRate)
          }
          .map { case ((courseName, classId), (present, absent, total, rate)) =>
            (courseName, classId, present, absent, total,
              BigDecimal(rate).setScale(2, BigDecimal.RoundingMode.HALF_UP).toDouble)
          }
          .sortBy(_._1) // 按课程名称排序

        // 5. 每个班级的学生出勤详细信息
        val studentAttendanceByClassRDD = attendanceRecordsRDD
          .map(record => ((record.classId, record.studentId, record.studentName), record.isAbsent))
          .groupByKey()
          .mapValues { absenceValues =>
            val valuesList = absenceValues.toList
            val totalCount = valuesList.size
            val absentCount = valuesList.count(_ == 1)
            val presentCount = totalCount - absentCount
            (presentCount, absentCount, totalCount)
          }
          .map { case ((classId, studentId, studentName), (present, absent, total)) =>
            (classId, studentId, studentName, present, absent, total)
          }
          .sortBy(x => (x._1, x._5), ascending = false) // 按班级和缺勤数排序

        // 打印班级出勤统计结果
        println("\n【各班级出勤统计】")
        println("+-----------+------------+------------+-------------+-------------+")
        println("| 班级号     | 出勤人数    | 缺勤人数    | 总人数       | 出勤率(%)    |")
        println("+-----------+------------+------------+-------------+-------------+")

        classAttendanceRDD.collect().foreach { summary =>
          println(f"| ${summary.classId}%-9s | ${summary.presentCount}%-10d | ${summary.absentCount}%-10d | ${summary.totalCount}%-11d | ${summary.attendanceRate}%-11.2f |")
        }

        println("+-----------+------------+------------+-------------+-------------+")

        // 打印总体出勤统计结果
        println("\n【总体出勤统计】")
        println("+-----------+------------+------------+-------------+-------------+")
        println("| 统计项     | 出勤人数    | 缺勤人数    | 总人数       | 出勤率(%)    |")
        println("+-----------+------------+------------+-------------+-------------+")
        println(f"| 所有班级   | ${totalPresent}%-10d | ${totalAbsent}%-10d | ${totalCount}%-11d | ${overallAttendanceRate}%-11.2f |")
        println("+-----------+------------+------------+-------------+-------------+")

        // 打印课程班级交叉统计
        println("\n【课程班级交叉出勤统计】")
        println("+-----------+----------+------------+------------+-------------+-------------+")
        println("| 课程名     | 班级号    | 出勤人数    | 缺勤人数    | 总人数       | 出勤率(%)    |")
        println("+-----------+----------+------------+------------+-------------+-------------+")

        coursesPerClassRDD.collect().foreach { case (course, classId, present, absent, total, rate) =>
          println(f"| ${course}%-9s | ${classId}%-8s | ${present}%-10d | ${absent}%-10d | ${total}%-11d | ${rate}%-11.2f |")
        }

        println("+-----------+----------+------------+------------+-------------+-------------+")

        // 打印班级学生详细出勤信息（仅显示每个班级前5名缺勤学生）
        println("\n【班级学生出勤详情】(每班显示前5名缺勤最多的学生)")

        // 使用RDD操作获取每个班级的学生列表并分别显示
        val classMostAbsentStudents = studentAttendanceByClassRDD
          .groupBy(_._1) // 按班级分组
          .mapValues(studentsIter => {
            // 取每个班级缺勤次数最多的前5名学生
            studentsIter.toList
              .sortBy(-_._5) // 按缺勤次数降序排序
              .take(5)
          })
          .collect()

        // 遍历每个班级
        classMostAbsentStudents.sortBy(_._1).foreach { case (classId, students) =>
          println(s"\n班级: $classId")
          println("+------------+------------------+------------+------------+-------------+")
          println("| 学号       | 学生姓名         | 出勤次数    | 缺勤次数    | 总课次       |")
          println("+------------+------------------+------------+------------+-------------+")

          students.foreach { case (_, studentId, studentName, present, absent, total) =>
            println(f"| ${studentId}%-10s | ${studentName}%-16s | ${present}%-10d | ${absent}%-10d | ${total}%-11d |")
          }

          println("+------------+------------------+------------+------------+-------------+")
        }

        // 保存到MySQL数据库
        try {
          // 1. 更新总体出勤统计
          updateAttendanceSummary(totalPresent, totalAbsent)

          // 2. 保存班级统计到数据库
          saveClassStatsToDb(classAttendanceRDD.collect())

          // 3. 记录可能缺勤学生
          savePotentialAbsentees(studentAttendanceByClassRDD.filter(_._5 > 0).collect())

          println("\n✓ 统计数据已保存到MySQL数据库")
        } catch {
          case e: Exception =>
            println(s"\n✗ 保存数据到MySQL时出错: ${e.getMessage}")
            e.printStackTrace()
        }

        // 提交偏移量
        stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
      }
    }

    // 启动Streaming Context
    ssc.start()
    println("\n✓ Spark RDD 班级出勤统计程序已启动")
    println("✓ 正在监听Kafka数据流...")
    ssc.awaitTermination()
  }

  /**
   * 更新总体出勤统计
   */
  def updateAttendanceSummary(presentCount: Int, absentCount: Int): Unit = {
    var connection: Connection = null
    var statement: PreparedStatement = null

    try {
      connection = createConnection()
      statement = connection.prepareStatement(
        """
          |INSERT INTO attendance_summary (total_present, total_absent, last_updated)
          |VALUES (?, ?, NOW())
          |ON DUPLICATE KEY UPDATE
          |total_present = total_present + VALUES(total_present),
          |total_absent = total_absent + VALUES(total_absent),
          |last_updated = NOW()
          |""".stripMargin)

      statement.setInt(1, presentCount)
      statement.setInt(2, absentCount)
      statement.executeUpdate()
    } finally {
      if (statement != null) statement.close()
      if (connection != null) connection.close()
    }
  }

  /**
   * 保存班级统计数据到MySQL
   */
  def saveClassStatsToDb(classStats: Array[ClassAttendanceSummary]): Unit = {
    var connection: Connection = null
    var statement: PreparedStatement = null

    try {
      connection = createConnection()
      statement = connection.prepareStatement(
        """
          |INSERT INTO class_attendance
          |(class_id, present_count, absent_count, last_updated)
          |VALUES (?, ?, ?, NOW())
          |ON DUPLICATE KEY UPDATE
          |present_count = present_count + VALUES(present_count),
          |absent_count = absent_count + VALUES(absent_count),
          |last_updated = NOW()
          |""".stripMargin)

      classStats.foreach { summary =>
        statement.setString(1, summary.classId)
        statement.setInt(2, summary.presentCount)
        statement.setInt(3, summary.absentCount)
        statement.addBatch()
      }

      statement.executeBatch()
    } finally {
      if (statement != null) statement.close()
      if (connection != null) connection.close()
    }
  }

  /**
   * 保存可能缺勤学生信息到缺勤预测表
   */
  def savePotentialAbsentees(absentees: Array[(String, String, String, Int, Int, Int)]): Unit = {
    var connection: Connection = null
    var statement: PreparedStatement = null

    try {
      connection = createConnection()
      statement = connection.prepareStatement(
        """
          |INSERT INTO absence_prediction
          |(student_id, course_name, probability, rank_order, last_updated)
          |VALUES (?, 'All', ?, ?, NOW())
          |ON DUPLICATE KEY UPDATE
          |probability = VALUES(probability),
          |rank_order = VALUES(rank_order),
          |last_updated = NOW()
          |""".stripMargin)

      // 按缺勤率排序
      val sortedAbsentees = absentees
        .map { case (_, studentId, _, _, absent, total) =>
          (studentId, (absent.toDouble / total) * 100)
        }
        .sortBy(-_._2) // 按缺勤率降序排列

      // 添加排序后的学生到数据库
      var rank = 1
      sortedAbsentees.foreach { case (studentId, probability) =>
        statement.setString(1, studentId)
        statement.setDouble(2, probability)
        statement.setInt(3, rank)
        statement.addBatch()
        rank += 1
      }

      statement.executeBatch()
    } finally {
      if (statement != null) statement.close()
      if (connection != null) connection.close()
    }
  }

  /**
   * 创建MySQL数据库连接
   */
  def createConnection(): Connection = {
    // 加载数据库驱动
    Class.forName("com.mysql.cj.jdbc.Driver")

    // 创建连接
    DriverManager.getConnection(
      "jdbc:mysql://43.143.125.94:3306/attendance_new?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai",
      "root",
      "Lwj378$$"
    )
  }
}